package org.wxd.mis.reg.api;

import org.apache.zookeeper.*;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.stereotype.Component;

import java.util.Objects;
import java.util.concurrent.CountDownLatch;

/**
 * 服务注册实现类
 * Created by wangxd on 2017/3/14.
 */
@Component
public class ServiceRegistryImpl implements ServiceRegistry, Watcher {

    private static final Logger logger = LoggerFactory.getLogger(ServiceRegistryImpl.class);

    private static final int SESSION_TIMEOUT = 5000;
    private static final String REGISTRY_PATH = "/registry";
    private static CountDownLatch latch = new CountDownLatch(1);

    private ZooKeeper zk;

    public ServiceRegistryImpl() {
    }

    public ServiceRegistryImpl(String zkServices) {
        try {
            zk = new ZooKeeper(zkServices, SESSION_TIMEOUT, this);
            latch.await();
            logger.info("connected to zk.the zkServices addr: {}.",zkServices);
        } catch (Exception e) {
            logger.error("create zk client error.the zkServices addr: {}", zkServices);
        }
    }

    public void registry(String serviceName, String serviceAddress) {
        try {
            // 根节点
            String regPath = REGISTRY_PATH;
            // 服务名称节点
            String servicePath = regPath + "/" + serviceName;

            createNode(regPath).createNode(servicePath);

            String addressPath = servicePath + "/" + serviceAddress;
            String addressNode = zk.create(addressPath, serviceAddress.getBytes(), ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.EPHEMERAL_SEQUENTIAL);
            logger.info("create addr node: {} => {}", addressNode, serviceAddress);
        } catch (Exception e) {
            logger.error("registry {} => {} error.",serviceName,serviceAddress);
        }

    }

    private ServiceRegistryImpl createNode(String path) throws KeeperException, InterruptedException {
        if (Objects.isNull(zk.exists(path, false))) {
            zk.create(path, null, ZooDefs.Ids.OPEN_ACL_UNSAFE, CreateMode.PERSISTENT);
            logger.info("create node: {}", path);
        }
        return this;
    }

    public void process(WatchedEvent watchedEvent) {
        if (watchedEvent.getState() == Event.KeeperState.SyncConnected) {
            // 连接上了放行
            latch.countDown();
        }

    }
}
